package consumer

import (
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	uuid "github.com/satori/go.uuid"
	"github.com/zeromicro/go-zero/core/logx"
	"go-zero-project/common/excption"
	"go-zero-project/services/user/api/constans"
	"go-zero-project/services/user/api/internal/config"
	"go-zero-project/services/user/api/internal/logic/msg"
	"go-zero-project/services/user/api/internal/svc"
)

// InitConsumer InitUniversalConsumer 消费 ROCKETMQ_GO_LOG_LEVEL 设置为error 不打印ping日志
func InitConsumer(config *config.MqConfig, serCtx *svc.ServiceContext) error {
	var groupName = config.GroupName
	namesrvs := []string{config.Namesrv}
	// 设置消息轨迹
	traceCfg := &primitive.TraceConfig{
		Access:   primitive.Local,
		Resolver: primitive.NewPassthroughResolver(namesrvs),
	}
	c, _ := rocketmq.NewPushConsumer(
		consumer.WithGroupName(groupName),
		consumer.WithConsumerModel(consumer.Clustering),
		consumer.WithNsResolver(primitive.NewPassthroughResolver(namesrvs)),
		consumer.WithTrace(traceCfg),
	)
	// 账号密码
	accessKey := config.AccessKey
	key := config.SecretKey
	if len(accessKey) > 0 && len(key) > 0 {
		credentials := primitive.Credentials{AccessKey: accessKey, SecretKey: key}
		consumer.WithCredentials(credentials)
	}
	topicList := config.TopicList
	if len(topicList) < 1 {
		return excption.NewErrCode(constans.MqSkipLoadErr)
	}
	instanceId := uuid.NewV4()
	consumer.WithInstance(fmt.Sprintf("%d", instanceId))
	for _, topicInfo := range topicList {
		// 设置Tag consumer.MessageSelector{Type: consumer.TAG, Expression: "user-api"}
		err := c.Subscribe(topicInfo.TopicName, consumer.MessageSelector{Type: consumer.TAG,
			Expression: topicInfo.TopicTag}, func(ctx context.Context,
			msgArr ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
			for i := range msgArr {
				messageExt := msgArr[i]
				consumerMsgLogic := msg.NewConsumerMsgLogic(ctx, serCtx)
				consumerMsgLogic.ConsumerMsg(messageExt)
			}
			return consumer.ConsumeSuccess, nil
		})
		if err != nil {
			logx.Errorf(err.Error())
		}
	}
	// Note: start after subscribe
	startErr := c.Start()
	if startErr != nil {
		logx.Errorf(startErr.Error())
		return startErr
	}
	return nil
}
